package com.gopay.scheduler.job;

import java.util.ArrayList;
import java.util.Calendar;
import java.util.Date;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

import javax.annotation.Resource;

import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.gopay.common.datasource.DBType;
import com.gopay.common.datasource.DbContextHolder;
import com.gopay.common.domain.report.Event;
import com.gopay.common.event.EventBus;
import com.gopay.common.event.EventState;
import com.gopay.common.report.dao.MqEventInfoDAO;
import com.gopay.common.report.dao.MqEventTimeoutLogDAO;
import com.gopay.common.util.DateUtils;
import com.gopay.scheduler.GopayBaseJob;
import com.gopay.scheduler.service.eventshandler.impl.EventPublishTask;

/**
 * 事件处理任务
 * 
 * @EventsHandlerJob.java
 * @author zhaoxubing
 * @2015年5月29日 下午4:35:08 www.gopay.com.cn Inc.All rights reserved.
 */
public class EventsHandlerJob extends GopayBaseJob {
    // 日志
    private Logger logger = LoggerFactory.getLogger(this.getClass());
    
    private static final int MAX_THREAD_COUNT=12;
    private static final int TOTAL_EVENT_COUNT=480;
    
    private ExecutorService threadPool = Executors.newFixedThreadPool(MAX_THREAD_COUNT);

    /**
     * 余数
     */
    private int remainder;
    
    private int modInt;
    
    @Resource(name="mqEventInfoDAO")
    private MqEventInfoDAO mqEventInfoDAO;

    @Resource(name="mqEventTimeoutLogDAO")
    private MqEventTimeoutLogDAO mqEventTimeoutLogDAO;

    @Resource(name = "eventBus")
    private EventBus eventBus;

    public void doJobInternal() {
        logger.info("EventsHandlerJob" + remainder + "------start------");
      
        List<Event> resultList = queryEvents();
       
        logger.info("resultList.size=" + resultList.size());
        if (null != resultList && !resultList.isEmpty()) {
           
            HashMap<String, List<Event>> sameOrderEventMap = new HashMap<String, List<Event>>();
            String key;
            List<Event> orderEventList;
            for (Event evt : resultList) {
                key = StringUtils.isNotBlank(evt.getBatchTaskId()) ? evt.getBatchTaskId() : evt.getGopayOrderId();
                orderEventList = sameOrderEventMap.get(key);
                if (orderEventList == null) {
                    orderEventList = new ArrayList<Event>();
                }
                orderEventList.add(evt);

                sameOrderEventMap.put(key, orderEventList);
            }

            try {
                List<List<Event>> taskEventList=new ArrayList<List<Event>>();                
                Iterator<String> iterator = sameOrderEventMap.keySet().iterator();
                for(int i=0;i<MAX_THREAD_COUNT&&iterator.hasNext();i++){
                    List<Event> tmpList=new ArrayList<Event>();                    
                    while(iterator.hasNext()&&tmpList.size()<TOTAL_EVENT_COUNT/MAX_THREAD_COUNT+1){
                        String tmpKey = iterator.next();
                        tmpList.addAll(sameOrderEventMap.get(tmpKey));
                    }
                    
                    taskEventList.add(tmpList);
                }
                
                //相同事件去重
                excludeSameEvent(taskEventList);
               
                // 同步锁存器
                CountDownLatch latch = new CountDownLatch(taskEventList.size());                
                for (List<Event> eList : taskEventList) {                    
                    EventPublishTask thread= new EventPublishTask(eventBus, eList, latch);
                    threadPool.execute(thread);
                }
                
                // 同步锁存器等待所有线程执行完成
                if(!latch.await(7, TimeUnit.SECONDS)){
                    logger.error("同步锁存器等待超时了 latch.count=" + latch.getCount());
                    threadPool.awaitTermination(1, TimeUnit.SECONDS);
                }

                

            } catch (Exception e) {
                logger.error("ExecutorService.execute ", e);
            }

            
            
        }
        
        logger.info("EventsHandlerJob" + remainder + "------end------");
    }

    private List<Event> queryEvents() {
        Calendar cal = Calendar.getInstance();
        cal.setTime(new Date());
        cal.add(Calendar.DAY_OF_YEAR, -7);

        Map<String, Object> map = new HashMap<String, Object>();
        map.put("startDate", DateUtils.format(cal.getTime(), "yyyyMMdd"));
        String queryHql = "FROM Event e WHERE (mod(e.gopayOrderId," + modInt + ") = " + remainder + " or mod(e.batchTaskId," + modInt + ") = " + remainder + ") and  e.createdTs >= TO_DATE(:startDate, 'YYYY-MM-DD HH24:mi:ss') and e.stat = '01' order by e.createdTs asc";
//        logger.info("queryHql=" + queryHql);
        DbContextHolder.setDbType(DBType.DS_REPORT.value);
        List<Event> resultList = mqEventInfoDAO.findByHql(queryHql, map, 1, TOTAL_EVENT_COUNT);
        
        return resultList;
    }
    
    /**
     * 排除相同事件
     * @param taskEventList
     * @return
     */
    private List<List<Event>> excludeSameEvent(List<List<Event>> taskEventList){
        for(List<Event> evtList:taskEventList){
            Map<String,Event> map=new HashMap<String,Event>();
            for(int i=0;i<evtList.size();){
                Event evt=evtList.get(i);
                if(StringUtils.isBlank(evt.getGopayOrderId())){
                    i++;
                    continue;
                }
                
                String key=evt.getGopayOrderId()+"_"+evt.getType();
                if(map.containsKey(key)){
                    //更新事件状态
                    evt.setStat(EventState.SUCCESS.value);
                    evt.setRemark1("dup");
                    updateEventStat(evt);
                    evtList.remove(i);
                }else{
                    map.put(key, evt);
                    i++;
                }
            }
        }
        
        return taskEventList;
    }
    
    private void updateEventStat(Event evt){
        DbContextHolder.setDbType(DBType.DS_REPORT.value);
        evt.setLastUpdTs(new Date());
        try{
            mqEventInfoDAO.update(evt);
        }catch(Exception e){
            logger.error("更新事件表出错 evt.id=" + evt.getId() + ", evt.gopayOrderId" + evt.getGopayOrderId(), e);
        }
    }
    
    public void setRemainder(int remainder) {
        this.remainder = remainder;
    }

    public void setModInt(int modInt) {
        this.modInt = modInt;
    }
    
    public int getSemaphore() {
        return modInt;
    }

}
